package org.jgroups.tests;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.jgroups.protocols.FRAG3;
import org.jgroups.protocols.FragHeader3;

import java.util.List;
import java.util.ArrayList;
import java.util.Vector;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.TimeUnit;
import java.io.File;
import static org.jgroups.tests.TestSender.*;
import org.jgroups.*;

/**
 * Created by IntelliJ IDEA.
 * User: thsieh
 * Date: Jul 14, 2009
 * Time: 9:26:40 AM
 * To change this template use File | Settings | File Templates.
 */
public class TestObjSender   extends ReceiverAdapter {
    private static Log logger = LogFactory.getLog(TestObjSender.class);

    private List<Long> begin = new ArrayList<Long>();
    private List<Long> current;
    AtomicInteger no ;
    int count;
    FRAG3 frag3;
    int load ;
    private long[] times = new long[4];
    public final String[] objectsLists = {"acamps", "urlfilters", "acamps", "advertisers", "aorders", "arkcreatives", "arkdata",
            "behaviors" ,"cpmroiadjustments", "creatives", "cyclelimitcare", "effectivecpm2_1", "effectivecpm2_2",
            "effectivecpm2_3","effectivecpm2_4", "effectivecpm2_5","effectivecpm3_1", "effectivecpm3_2", "effectivecpm3_3",
            "effectivecpm3_4", "effectivecpm3_5", "filtereligibility",   "loc2pcamp", "locations","pcamps", "urlfilters" };

    public final String[] rules = { "effectivecpm2_1", "effectivecpm2_2",
        "effectivecpm2_3","effectivecpm2_4", "effectivecpm2_5","effectivecpm3_1", "effectivecpm3_2", "effectivecpm3_3",
        "effectivecpm3_4", "effectivecpm3_5", "arkcreatives", "arkdata"
        };
    public final String[] caps= { "config", "acampsbypanel", "cyclelimitcare", "cpmroiadjustments", "filtereligibility"
        };
    public final String[] data = { "acamp2adv", "acamps", "adtypes", "pcamps" , "advertisers" , "aorders", "creatives",
        "urlfilters", "loc2pcamp", "locations" };
    public final String[] misc = { "behaviors" ,"zipcodes", "demographic", "iptocountry", "timebasedwearoutfactor",
        "iptocountry", "demographic", "timebasedwearoutfactor", "browserinfomapping"
        };

    // default of interval cap 1 min, data 4 mins, rules 6 mins, misc 30 mins
    // in seconds
    public final long[] Interval = { 1*60 , 4 * 60, 6 * 60 , 30 * 60  };
    private String path ;
    private JChannel channel;
    private int size = 1;

    public TestObjSender(int count, String path) {
        this.count = count ;
        this.path = path ;
        this.no = new AtomicInteger(0);
        List<String> list = runShell(new String[] { "sh","-c","netstat -s | grep -A4 Udp" }) ;
        for ( String each : list) {
            String[] strs = each.split("\\s");
            begin.add( Long.valueOf( strs[0]) );
        }
        current = begin;
        logger.info( output(begin));
        for (  int i = 0; i < Interval.length ; i ++)  times[i] = System.currentTimeMillis();
    }

    public void receive(Message msg) {
        if ( msg == null ) {
            logger.warn("Mesasge is null ");
        }
        else {
            FragHeader3 header = (FragHeader3) msg.getHeader("FRAG3") ;
            logger.info("Receive message source=" +msg.getSrc().toString()+" length="+ msg.getLength()
                    +" destination="+ msg.getDest()+" "+ ( header == null ? "null" : header.toString()));
        }
        int i = no.getAndIncrement();
        if ( i % count == (count -1 ) ) {
            List<Long> now =  new ArrayList<Long>();
            logger.info("FRAG3 " + frag3.toString() );
            List<String> list = runShell(new String[] { "sh","-c","netstat -s | grep -A4 Udp" }) ;
            for ( String each : list) {
                String[] strs = each.split("\\s");
                now.add( Long.valueOf( strs[0]) );
            }
            long dif = now.get(2) - current.get(2);
            if ( dif == 0 ) logger.info(i+" messages no udp error");
            else logger.info(i+" messages extra udp error " + dif);
            current = now;
        }
    }

    public void printUpd() {
        List<Long> now =  new ArrayList<Long>();
        logger.info("FRAG3 " + frag3.toString() );
        List<String> list = runShell(new String[] { "sh","-c","netstat -s | grep -A4 Udp" }) ;
        if ( list.size() < 2 ) return;
        for ( String each : list) {
            String[] strs = each.split("\\s");
            now.add( Long.valueOf( strs[0]) );
        }
        long dif = now.get(2) - current.get(2);
        if ( dif == 0 ) logger.info(" messages no udp error");
        else logger.info(" messages extra udp error " + dif);
        current = now;
    }

    public static void help() {
          System.out.println("java stress.org.jgroups.test.TestSender -udp ./udp.xml -cluster test-1 -count 100 size -size 10");
    }

    public static boolean isInLists(String name, String[] names) {
        for (String each : names) {
            if ( name.indexOf( each) >= 0 ) return true;
        }
        return false;
    }

    public static List<File> findFiles(String path, String[] names) {
        File file = new File( path);
        File[] files = file.listFiles();
        List<File> list = new ArrayList<File>();
        for ( File each : files) {
           if ( isInLists(each.getName(), names)) list.add(each );
        }
        return list;
    }

    public byte[] getBytes(File file) {
        long size = file.length() * ( load >=1 ? load : 1);
        return new byte[ (int) size] ;
    }

    private String[] getCategory(int category) {
        switch ( category) {
            case 0 : return caps;
            case 1 : return data;
            case 2 : return rules;
            case 3 : return misc;
            default : return null;
        }
    }

    public void send(int category) {
        List<File> files = findFiles(path,  getCategory(category));
        for ( File each : files) {
            Message msg = new Message(null, null, getBytes(each) );
            try {
                channel.send( msg);
            } catch (Exception ex) {
                throw new RuntimeException( ex.getMessage(), ex);
            }
        }
    }

    public void send() {
        for ( int i =0 ; i < Interval.length ; i ++) {
            if ( (System.currentTimeMillis() - times[i] ) > Interval[i] * (1000/size) ) {
                send(i);
                times[i] = System.currentTimeMillis();
            }
        }
    }

    @Override
    public void viewAccepted(View view) {
            logger.info("ViewAccepted size=>"+ view.size()+"  "+view.printDetails() );
    }


    public static void main(String[] args) throws Exception {
        String[] defaults = { "./udp.xml","test-1" , "100" , "1", "2000", "3", "../datas" ,"1"};

        for ( int i=0; i < args.length ; ) {
            if ( args[i].equals("-udp")) defaults[0] = args[++i] ;
            else if ( args[i].equals("-cluster")) defaults[1] = args[++i] ;
            else if ( args[i].equals("-count")) defaults[2] = args[++i] ;
            else if ( args[i].equals("-size")) defaults[3] = args[++i] ;
            else if ( args[i].equals("-times")) defaults[4] = args[++i] ;
            else if ( args[i].equals("-sleep")) defaults[5] = args[++i] ;
            else if ( args[i].equals("-dir")) defaults[6] = args[++i] ;
            else if ( args[i].equals("-load")) defaults[7] = args[++i] ;
            else if ( args[i].equals("-help")) help() ;
            else {
               help();
            }
            i ++;
        }
        int c =  Integer.valueOf( defaults[2]);
        TestObjSender ts = new TestObjSender( c, defaults[6]);
        System.setProperty("java.net.preferIPv4Stack", "true");
        logger.info("java.net.preferIPv4Stack ="+ System.getProperty("java.net.preferIPv4Stack"));

        JChannel channel = new JChannel(new File(defaults[0]));
        channel.setReceiver(ts);
        channel.connect(defaults[1]);
        int count = Integer.valueOf( defaults[2]);
        int size  = Integer.valueOf( defaults[3]);
        int times =  Integer.valueOf( defaults[4]);
        int sleep = Integer.valueOf( defaults[5]);
        int load =  Integer.valueOf( defaults[7]);
        ts.size = size;
        ts.channel = channel;
        ts.load = load ;
        ts.frag3 = (FRAG3) channel.getProtocolStack().getTopProtocol();
        for (int j= 0; j < times ; j ++ ) {
            logger.info("Begin iteration " + j );
            long begin = System.nanoTime();
            ts.send();
            Thread.sleep( 1000L * sleep );
            logger.info("FRAG3 "+ ts.frag3.toString());
            ts.printUpd();
        }

    }


}
